#if defined(MODEL_ROOMS)
#include "kafkaService.h"
#include <iostream>
#include <string>
#include <thread>
#include "rdkafkacpp.h"
#include "rdkafka.h"
#include "json.h"
#include "log.h"
#include "ConfigApi.h"


using namespace ConfigApi;

namespace Media{

kafkaService *kafkaService::sm_deviceInfoInstance = nullptr;
mutex      kafkaService::sm_instanceLock;

kafkaService *kafkaService::instance()
{
    std::lock_guard<mutex> guard(sm_instanceLock);
    
	if(sm_deviceInfoInstance == nullptr)
	{
		sm_deviceInfoInstance = new kafkaService();
	}

	return sm_deviceInfoInstance;
}


bool kafkaService::startReportThd(){
    if(!m_reportExit){
        m_reportMark = false;
        while(!m_reportExit){
            sleep(1);
        }
    }
    m_reportMark = true;
    auto lam = [this](){
        while(m_reportMark){
            #if 1
            //获取kafka的brokers和topic信息
            Json::Value kafkaTable;
            ConfigApi::IConfigInterface::GetInstance()->Get("Kafka", kafkaTable);
            //判断当前是否存在
            if(kafkaTable.isNull() || !kafkaTable.isMember("brokers") || !kafkaTable.isMember("topic") || !kafkaTable.isMember("username") || !kafkaTable.isMember("password")){
                continue;
            }	
            string brokers  =  kafkaTable["brokers"].asString();
            string topic	= kafkaTable["topic"].asString();
            string username = kafkaTable["username"].asString();
            string password = kafkaTable["password"].asString();
            #endif
            DeviceInfo devInfo;
            if(devInfo.getDevInfo()){
                if(!devInfo.KafkaInfoCompare(m_curDevInfo)){
                    Json::Value table;
                    devInfo.getkafkaJson(table);
                    LOG_PrintInfo("json ------> %s\n", table.toStyledString().c_str());
                    reportMessage(brokers, topic, table.toStyledString(), username, password);
                    m_curDevInfo = devInfo;
                }
            }else{
                LOG_PrintWarn("faild to get device info\n");
            }
            sleep(3);
        }
        m_reportExit = true;
    };
    thread reportThd(lam);
    reportThd.detach();
    return true;
}


class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) {
        if (message.err()) {
            std::cerr << "消息发送失败: " << message.errstr() << std::endl;
        } else {
            std::cout << "消息已发送到主题: " << message.topic_name() << ", 分区: " << message.partition() << ", 偏移量: " << message.offset() << std::endl;
        }
    }
};

bool kafkaService::reportMessage(string brokers, string topic, string message, string username, string password) {
    int version = rd_kafka_version();
    const char* version_str = rd_kafka_version_str();

    std::cout << "librdkafka version: " << version_str << " (" << version << ")" << std::endl;
    int32_t partition = RdKafka::Topic::PARTITION_UA;
    int message_count = 1;

    ExampleDeliveryReportCb delivery_report_callback;

    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    std::string errstr;

    if (conf->set("bootstrap.servers", brokers, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "Error: 设置broker列表失败 - " << errstr << std::endl;
        delete conf;
        return false;
    }

  
    // 不启用 SASL_SSL 安全协议
    if (conf->set("security.protocol", "SASL_PLAINTEXT", errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "Error: 启用 SASL_SSL 失败 - " << errstr << std::endl;
        delete conf;
        return false;
    }
    
    

    // 设置 SASL 机制为 SCRAM-SHA-256
    if (conf->set("sasl.mechanism", "SCRAM-SHA-256", errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "Error: 设置 SASL 机制失败 - " << errstr << std::endl;
        delete conf;
        return false;
    }



    // 设置用户名和密码
    if (conf->set("sasl.username", username, errstr) != RdKafka::Conf::CONF_OK ||
        conf->set("sasl.password", password, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << "Error: 设置用户名和密码失败 - " << errstr << std::endl;
        delete conf;
        return false;
    }

    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Error: 创建生产者失败 - " << errstr << std::endl;
        delete conf;
        return false;
    }

    RdKafka::Topic *topic_ptr = RdKafka::Topic::create(producer, topic, NULL, errstr);
    if (!topic_ptr) {
        std::cerr << "Error: 创建主题对象失败 - " << errstr << std::endl;
        delete producer;
        delete conf;
        return false;
    }

    for (int i = 0; i < message_count; i++) {
        std::string msg = message;
        RdKafka::ErrorCode resp = producer->produce(topic_ptr, partition, RdKafka::Producer::RK_MSG_COPY, const_cast<char *>(msg.c_str()), msg.size(), NULL, NULL);
        if (resp != RdKafka::ERR_NO_ERROR) {
            std::cerr << "Error: 发送消息失败 - " << RdKafka::err2str(resp) << std::endl;
            delete topic_ptr;
            delete producer;
            delete conf;
            return false;
        }

        producer->poll(0);
    }

    producer->flush(100 * 1000);

    delete topic_ptr;
    delete producer;
    delete conf;

    return true;
}
}

#endif